package consumer_test

import (
	"context"
	"gitee.com/zaper/kafka-service/model"
	"gitee.com/zaper/private-logger"
	"github.com/IBM/sarama"
	"github.com/petermattis/goid"
	"time"
)

type ConsumerGroupHandlerImpl struct {
	GroupId string
}

func NewConsumerGroupHandlerImpl(groupId string) *ConsumerGroupHandlerImpl {
	return &ConsumerGroupHandlerImpl{
		GroupId: groupId,
	}
}

func (consumerGroupHandlerImpl *ConsumerGroupHandlerImpl) Setup(sarama.ConsumerGroupSession) error {
	return nil
}

func (consumerGroupHandlerImpl *ConsumerGroupHandlerImpl) Cleanup(session sarama.ConsumerGroupSession) error {
	return nil
}

func (consumerGroupHandlerImpl *ConsumerGroupHandlerImpl) ConsumeClaim(sess sarama.ConsumerGroupSession,
	claim sarama.ConsumerGroupClaim) error {
	for msg := range claim.Messages() {
		logger.Infof("消费者：主题=%s | 分区=%d | 偏移量=%d | 消费组=%s | %d号消费者读取到：%s",
			msg.Topic, msg.Partition, msg.Offset, consumerGroupHandlerImpl.GroupId, goid.Get(), string(msg.Value))
		sess.MarkMessage(msg, "") // 标记消息已处理
		sess.Commit()
	}
	return nil
}

func execConsumerMess(kafkaModel *model.KafkaConsumerModel) {
	logger.Infof("%d号消费者进入%s消费组", goid.Get(), kafkaModel.GroupId)
	brokerAdds := []string{"localhost:9091"}
	config := sarama.NewConfig()
	config.Consumer.Return.Errors = true
	config.Consumer.Offsets.Initial = sarama.OffsetNewest
	// 2. 创建消费者组
	group, err := sarama.NewConsumerGroup(brokerAdds, kafkaModel.GroupId, config)
	if err != nil {
		logger.Infof("创建消费者组失败: %v", err)
		return
	}
	defer func() {
		err := group.Close()
		if err != nil {

		}
	}()
	defer close(kafkaModel.StopChan[goid.Get()])
	// 3. 消费消息（匿名处理器）
	handler := NewConsumerGroupHandlerImpl(kafkaModel.GroupId)
	ctx := context.Background()
	for {
		select {
		case <-kafkaModel.StopChan[goid.Get()]:
			logger.Info("kafka ibm custom shutting down")
			return
		default:
		}
		if err := group.Consume(ctx, kafkaModel.Topics, handler); err != nil {
			logger.Infof("消费错误: %v（200ms后重试）", err)
			time.Sleep(200 * time.Millisecond)
		}
	}
}
